package com.lgj.rocketmq.code;

import com.alibaba.fastjson.JSONObject;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.spring.support.RocketMQListenerContainer;
import org.springframework.beans.BeansException;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.context.SmartLifecycle;

import java.util.Map;

@Slf4j
public class ConsumerRocketMQListenerContainer implements RocketMQListenerContainer, SmartLifecycle, ApplicationContextAware {

    private volatile boolean running = false;

    private MQPushConsumerFactory mqPushConsumerFactory;

    public ConsumerRocketMQListenerContainer(MQPushConsumerFactory mqPushConsumerFactory){
        this.mqPushConsumerFactory = mqPushConsumerFactory;
    }

    private ApplicationContext applicationContext;

    @Override
    public void destroy() throws Exception {
        this.stop();
    }

    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        this.applicationContext = applicationContext;
    }

    @Override
    public void start() {
        Map<String, DefaultMQPushConsumer> consumerMap = mqPushConsumerFactory.getMQPushConsumers();
        consumerMap.forEach((topic,consumer)->{
            try {
                consumer.start();
                log.info("消费者【{}】启动成功！", JSONObject.toJSONString(consumer.getInstanceName()));
            } catch (MQClientException e) {
                throw new RuntimeException(e);
            }
        });
        running = true;
    }

    @Override
    public void stop() {
        Map<String, DefaultMQPushConsumer> consumerMap = mqPushConsumerFactory.getMQPushConsumers();
        consumerMap.forEach((topic,consumer)->{
            try {
                consumer.shutdown();
            } catch (Exception e) {
                throw new RuntimeException(e);
            }
        });
        running = false;
    }

    @Override
    public boolean isRunning() {
        return running;
    }
}
